package com.atguigu.chapter07;

import com.atguigu.chapter05.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/6 9:12
 */
public class Flink06_WindowFunction_Aggregate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        // 切分
                        String[] line = value.split(",");
                        return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));

                    }
                });


        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());

        WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorKS
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        // TODO 窗口的 增量聚合 函数
        sensorWS
                .aggregate(
                        new MyAggFunction(),
                        new MyProcessWindowFunction()
                )
                .print();


        env.execute();
    }

    public static class MyAggFunction implements AggregateFunction<WaterSensor, Long, Long> {

        @Override
        public Long createAccumulator() {
            System.out.println("create Acc...");
            return 0L;
        }

        @Override
        public Long add(WaterSensor value, Long accumulator) {
            System.out.println("add ...");
            return accumulator + value.getVc();
        }

        @Override
        public Long getResult(Long accumulator) {
            System.out.println("get ...");
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return null;
        }
    }

    public static class MyProcessWindowFunction extends ProcessWindowFunction<Long, String, String, TimeWindow> {

        @Override
        public void process(String key, Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
            out.collect("窗口:[" + context.window().getStart() + "," + context.window().getEnd() + ")\n" +
                    "当前key:" + key + "\n" +
                    "数据:" + elements);
        }
    }
}
/*
    aggregate（AggregateFunction，ProcessWindowFunction）
        第一个函数：进行增量聚合，并将 聚合结果 传递给 第二个函数
        第二个函数：全窗口函数，接收第一个函数的结果，进行进一步的处理

    好处： 兼顾了 增量的特点，和全窗口函数的 灵活性
 */